package com.ptkc.core.canal.handler;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.protobuf.InvalidProtocolBufferException;
import com.ptkc.utils.DataConvert;
import com.ptkc.utils.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;

import java.io.IOException;
import java.lang.reflect.Method;
import java.util.*;

/**
 * @author: Mr.Jiao
 * @date: 2018年09月30日
 */
@Slf4j
public abstract class EventTypeHandler<T> extends AbstractEventHandler {

    public EventTypeHandler(String database, String table) {
        super(database, table);
    }

    public abstract Class<? extends T> getEntryClass();

    @Override
    public boolean getSubscribe(String database, String table) {
        String _database = this.getDatabase();
        String _table = this.getTable();
        return (_database == null || _database.equalsIgnoreCase(database)) && StringUtil.isNotEmpty(_table) && _table.equalsIgnoreCase(table);
    }


    @Override
    public void handle(CanalEntry.Entry entry, CanalEntry.RowData rowData) {
        CanalEntry.EventType eventType = entry.getHeader().getEventType();
        T t = this.getRows(rowData, eventType);
        switch (eventType) {
            case INSERT:
                this.onInsert(t);
                break;
            case UPDATE:
                this.onUpdate(t);
                break;
            case DELETE:
                this.onDelete(t);
                break;
            default:
                this.onOtherType(t, eventType);
                break;
        }
    }

    public T getRows(CanalEntry.RowData rowData,CanalEntry.EventType eventType){
        List<CanalEntry.Column> columns = null;
        if(eventType == CanalEntry.EventType.DELETE){
            columns = rowData.getBeforeColumnsList();
        }else{
            columns = rowData.getAfterColumnsList();
        }
        Map<String, String> map = EventTypeHandler.this.getRowMap(columns);
        return DataConvert.convert(map, EventTypeHandler.this.getEntryClass());
    }

    public void onInsert(T rowData){}

    public void onUpdate(T rowData){}

    public void onDelete(T rowData){}

    public void onOtherType(T rowData, CanalEntry.EventType eventType){}
}